package com.avicsafety.ezsystem.rxjava;

import android.content.Context;
import android.support.v7.app.AppCompatActivity;
import android.os.Bundle;
import android.util.Log;
import android.view.View;
import android.widget.Button;
import android.widget.TextView;

import com.avicsafety.ezsystem.R;
import com.avicsafety.ezsystem.entity.cate;
import com.avicsafety.ezsystem.utils.SPUtils;
import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;

import java.lang.reflect.Type;
import java.util.List;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.annotations.NonNull;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;
import okhttp3.Call;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.Response;
import okhttp3.ResponseBody;


public class RxJavaActivity extends AppCompatActivity {

    private Button btn_base;
    private Button btn_thread;
    private Button btn_map;
    private Button btn_map2;
    private Button btn_concat;
    private Button btn_flat_map;

    private Context othis;

    private TextView mRxOperatorsText;

    private final String TAG = "MainActivity";

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_rx_java);

        othis = this;

        mRxOperatorsText = findViewById(R.id.mRxOperatorsText);

        btn_base = findViewById(R.id.btn_base);
        btn_base.setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View view) {
                baseFun();
            }
        });

        btn_thread = findViewById(R.id.btn_thread);
        btn_thread.setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View view) {
                threadFun();
            }
        });

        btn_map = findViewById(R.id.btn_map);
        btn_map.setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View view) {
                mapFun();
            }
        });

        btn_map2 = findViewById(R.id.btn_map2);
        btn_map2.setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View view) {
                map2Fun();
            }
        });

        btn_concat = findViewById(R.id.btn_concat);
        btn_concat.setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View view) {
                concatFun();
            }
        });

        btn_flat_map = findViewById(R.id.btn_flat_map);
        btn_flat_map.setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View view) {
                flatMapFun();
            }
        });



    }


    /**
     * 基础调用
     */
    private void baseFun() {
        Observable.create(new ObservableOnSubscribe<Integer>() { // 第一步：初始化Observable
            @Override
            public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
                Log.e(TAG, "Observable emit 1" + "\n");
                e.onNext(1);
                Log.e(TAG, "Observable emit 2" + "\n");
                e.onNext(2);
                Log.e(TAG, "Observable emit 3" + "\n");
                e.onNext(3);
                e.onComplete();  //完成后 Observer 不会接受到onNext()
                Log.e(TAG, "Observable emit 4" + "\n" );
                e.onNext(4);
            }
        }).subscribe(new Observer<Integer>() { // 第三步：订阅

            // 第二步：初始化Observer

            @Override
            public void onSubscribe(@NonNull Disposable d) {
                Log.e(TAG, "onSubscribe");
            }

            @Override
            public void onNext(@NonNull Integer integer) {
                mRxOperatorsText.setText(mRxOperatorsText.getText()+" onNext");
                Log.e(TAG, "onNext");
            }

            @Override
            public void onError(@NonNull Throwable e) {
                mRxOperatorsText.setText(mRxOperatorsText.getText()+"onError : value : " + e.getMessage());
                Log.e(TAG, "onError : value : " + e.getMessage() + "\n" );
            }

            @Override
            public void onComplete() {
                Log.e(TAG, "onComplete" + "\n" );
            }
        });

    }

    /**
     * 使用线程
     */
    private void threadFun() {

        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
                Log.e(TAG,"开始发射");
                Log.e(TAG, "Observable thread is : " + Thread.currentThread().getName());
                e.onNext(1);
                e.onNext(2);
                e.onComplete();
            }
        }).subscribeOn(Schedulers.newThread())
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .doOnNext(new Consumer<Integer>() {
                    @Override
                    public void accept(@NonNull Integer integer) throws Exception {
                        Log.e(TAG,"onNext 时执行");
                        Log.e(TAG, "After observeOn(mainThread)，Current thread is " + Thread.currentThread().getName());
                    }
                })
                .observeOn(Schedulers.io())
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(@NonNull Integer integer) throws Exception {
                        Log.e(TAG, "在IO线程执行");
                        Log.e(TAG, "After observeOn(io)，Current thread is " + Thread.currentThread().getName());
                    }
                });
    }

    /**
     * Map 操作符
     */
    private void mapFun() {

        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
                e.onNext(1);
                e.onNext(2);
                e.onNext(3);
            }
        }).map(new Function<Integer, String>() { //讲被订阅者的泛型改变 Integer - String
            @Override
            public String apply(@NonNull Integer integer) throws Exception {
                return "This is result " + integer;
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(@NonNull String s) throws Exception {
                Log.e(TAG, "accept : " + s +"\n" );
            }
        });

    }

    /**
     * 使用Map 操作符 转换Observable 为 一个实例
     */
    private void map2Fun() {

        Observable.create(new ObservableOnSubscribe<Response>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<Response> e) throws Exception {
                Log.e(TAG, "开始发射");
                Log.e(TAG, "Current thread is " + Thread.currentThread().getName());
                Request.Builder builder = new Request.Builder()
                        .url("https://ps.haishengshop.com/public/goods_cate/tree")
                        .get();
                Request request = builder.build();
                Call call = new OkHttpClient().newCall(request);
                Response response = call.execute();
                e.onNext(response);
            }
        }).map(new Function<Response, List<cate>>() {
            @Override
            public List<cate> apply(@NonNull Response response) throws Exception {
                if (response.isSuccessful()) {
                    ResponseBody body = response.body();
                    if (body != null) {
                        Log.e(TAG, "转换:" + response.body());
                        Log.e(TAG, "Current thread is " + Thread.currentThread().getName());
                        Type listType = new TypeToken<List<cate>>() {}.getType();
                        return new Gson().fromJson(body.string(), listType);
                    }
                }
                return null;
            }
        }).subscribeOn(Schedulers.io())
            .observeOn(Schedulers.newThread())
            .doOnNext(new Consumer<List<cate>>() {
                @Override
                public void accept(@NonNull List<cate> s) throws Exception {
                    Log.e(TAG, "doOnNext 在订阅者工作前执行：" + s.size() + "\n");
                    Log.e(TAG, "Current thread is " + Thread.currentThread().getName());
                }
            })
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Consumer<List<cate>>() {
                @Override
                public void accept(@NonNull List<cate> s) throws Exception {
                    Log.e(TAG, "订阅者执行 一般更新UI : " + s +"\n" );
                    Log.e(TAG, "Current thread is " + Thread.currentThread().getName());
                }
            });

    }

    /**
     * 使用concat 操作符
     */
    private void concatFun() {

        Observable<String> cache = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<String> e)  {
                Log.e(TAG, "create当前线程:"+Thread.currentThread().getName() );

                String data = (String) SPUtils.get(othis,"data_001","");

                // 在操作符 concat 中，只有调用 onComplete 之后才会执行下一个 Observable
                if (!data.equals("")){ // 如果缓存数据不为空，则直接读取缓存数据，而不读取网络数据
                    Log.e(TAG, "通过本地获取缓存数据:"+Thread.currentThread().getName() );
                    e.onNext(data);
                }else {
                    Log.e(TAG, "本地缓存没有数据:"+Thread.currentThread().getName() );
                    e.onComplete();
                }


            }
        });

        //访问网络的Observable
        Observable<String> network = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {
                Log.e(TAG, "开始通过网络获取");
                Log.e(TAG, "Current thread is " + Thread.currentThread().getName());
                Request.Builder builder = new Request.Builder()
                        .url("https://ps.haishengshop.com/public/goods_cate/tree")
                        .get();
                Request request = builder.build();
                Call call = new OkHttpClient().newCall(request);
                Response response = call.execute();
                e.onNext(response.body().string());
            }
        });


        // 两个 Observable 的泛型应当保持一致
        Observable.concat(cache,network)
                .subscribeOn(Schedulers.io())  //发射者的进程
                .observeOn(Schedulers.newThread())
                .doOnNext(new Consumer<String>() {
                    @Override
                    public void accept(@NonNull String s) throws Exception {
                        SPUtils.put(othis,"data_001",s);
                        Log.e(TAG, "显示数据之前 存储在本地:"+s );
                        Log.e(TAG, "显示数据之前 存储在本地:"+Thread.currentThread().getName() );
                    }
                })
                .observeOn(AndroidSchedulers.mainThread()) //订阅者的进程
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(@NonNull String data) throws Exception {
                        Log.e(TAG, "显示数据部分代码 主线程执行 subscribe 成功:"+Thread.currentThread().getName() );
                        mRxOperatorsText.append("accept: 读取数据成功");
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(@NonNull Throwable throwable) throws Exception {
                        Log.e(TAG, "subscribe 失败:"+Thread.currentThread().getName() );
                        Log.e(TAG, "accept: 读取数据失败："+throwable.getMessage() );
                        mRxOperatorsText.append("accept: 读取数据失败："+throwable.getMessage()+"\n");
                    }
                });

    }


    /**
     * 使用Flatmap 链式访问网络
     */
    private void flatMapFun() {


        //访问网络的Observable
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {
                Log.e(TAG, "开始通过网络获取 haishengshop");
                Log.e(TAG, "Current thread is " + Thread.currentThread().getName());
                Request.Builder builder = new Request.Builder()
                        .url("https://ps.haishengshop.com/public/goods_cate/tree")
                        .get();
                Request request = builder.build();
                Call call = new OkHttpClient().newCall(request);
                Response response = call.execute();
                e.onNext(response.body().string());
            }
        })
        .subscribeOn(Schedulers.io())        // 在io线程进行网络请求
        .observeOn(AndroidSchedulers.mainThread()) // 在主线程处理获取食品列表的请求结果
        .doOnNext(new Consumer<String>() {
            @Override
            public void accept(@NonNull String s) throws Exception {
                // 先根据获取食品列表的响应结果做一些操作
                Log.e(TAG, "accept: doOnNext :" + s.toString());
            }
        })
        .observeOn(Schedulers.io()) // 回到 io 线程去处理获取食品详情的请求
        .flatMap(new Function<String, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(@NonNull String foodList) throws Exception {
                return Observable.create(new ObservableOnSubscribe<String>() {
                    @Override
                    public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {
                        Log.e(TAG, "开始通过网络获取 avicsafety");
                        Log.e(TAG, "Current thread is " + Thread.currentThread().getName());
                        Request.Builder builder = new Request.Builder()
                                .url("http://www.avicsafety.com/ip_search.aspx")
                                .get();
                        Request request = builder.build();
                        Call call = new OkHttpClient().newCall(request);
                        Response response = call.execute();
                        e.onNext(response.body().string());
                    }
                });
            }
        })
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Consumer<String>() {
            @Override
            public void accept(@NonNull String foodDetail) throws Exception {
                Log.e(TAG, "accept: success ：" + foodDetail.toString());
                mRxOperatorsText.append("accept: success ：" + foodDetail.toString()+"\n");
            }
        }, new Consumer<Throwable>() {
            @Override
            public void accept(@NonNull Throwable throwable) throws Exception {
                Log.e(TAG, "accept: error :" + throwable.getMessage());
                mRxOperatorsText.append("accept: error :" + throwable.getMessage()+"\n");
            }
        });

    }

}
